package link.jfire.jnet.server.storage;

import link.jfire.baseutil.simplelog.ConsoleLogFactory;
import link.jfire.baseutil.simplelog.Logger;
import link.jfire.jnet.common.buffer.ByteBuf;
import link.jfire.jnet.common.exception.ServerStopException;
import link.jfire.jnet.common.handler.DataHandler;
import link.jfire.jnet.server.storage.ringarray.RingArray;

/**
 * 
 * entry的线程处理载体
 * 
 * @author 林斌
 *         
 */
public final class EntryAction implements Runnable
{
    // 当前准备处理的序号
    private volatile long   cursor = 0;
    private Logger          logger = ConsoleLogFactory.getLogger();
    private final RingArray ringArray;
                            
    public EntryAction(RingArray ringArray)
    {
        this.ringArray = ringArray;
    }
    
    @Override
    public void run()
    {
        Entry entry;
        while (true)
        {
            if (ringArray.isAvailable(cursor) == false)
            {
                try
                {
                    logger.debug("等待序号:{}", cursor);
                    ringArray.waitFor(cursor);
                }
                catch (ServerStopException e)
                {
                    logger.error("停止");
                    break;
                }
            }
            entry = ringArray.entryAt(cursor);
            if (entry.take() == false)
            {
                cursor += 1;
                continue;
            }
            ServerInternalResult result = entry.getInternalResult();
            try
            {
                if (result.getChannelInfo().isOpen() == false)
                {
                    logger.debug("通道{}已经关闭，不处理消息，直接退出", result.getChannelInfo().getAddress());
                    cursor += 1;
                    continue;
                }
                // 储存中间结果
                Object intermediateResult = result.getData();
                DataHandler[] handlers = result.getChannelInfo().getHandlers();
                for (int i = result.getIndex(); i < handlers.length;)
                {
                    intermediateResult = handlers[i].handle(intermediateResult, result);
                    if (i == result.getIndex())
                    {
                        i++;
                        result.setIndex(i);
                    }
                    else
                    {
                        i = result.getIndex();
                    }
                }
                if (intermediateResult instanceof ByteBuf<?>)
                {
                    result.setData(intermediateResult);
                    result.flowDone();
                    if (result.getChannelInfo().isOpen())
                    {
                        result.getChannelInfo().write(result);
                    }
                }
                else
                {
                
                }
                cursor += 1;
            }
            catch (Exception e)
            {
                logger.error("消息处理过程发生异常，关闭通道", e);
                result.getChannelInfo().close();
            }
        }
    }
    
    public long cursor()
    {
        return cursor;
    }
}
